package spring.cloud.tasks.client;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.context.ApplicationContext;
import spring.cloud.tasks.Worker;
import spring.cloud.tasks.client.item_task_processor.*;
import spring.cloud.tasks.client.item_task_processor.task_items_processor.task_data_processor.TaskDataProcessor;
import spring.cloud.tasks.client.util.Addresss;
import spring.cloud.tasks.common.utils.ApplicationContextUtils;
import spring.cloud.tasks.common.zookeeper.ZooKeeperItemTaskDataManager;
import spring.cloud.tasks.common.zookeeper.ZooKeeperManager;

import java.io.FileNotFoundException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

//import org.springframework.boot.tasks.client.task_batch_processor.BatchTaskProcessor;
//import org.springframework.boot.tasks.client.task_batch_processor.WorkerTaskBatchInfo;

@Slf4j
public class ItemTaskWorker extends Worker {


    @Setter
    @Getter
    private String workerId;
    @Setter
    @Getter
    private String ip;
    @Setter
    @Getter
    private String hostName;
    @Setter
    @Getter
    protected ZooKeeperManager zooKeeperManager;
    @Setter
    @Getter
    private ZooKeeperItemTaskDataManager zooKeeperDataManager;

    @Setter
    @Getter
    private ZooKeepersConnectAndWorkerTasksInitThread zooKeepersConnectAndWorkerTasksInitThread;
    @Setter
    @Getter
    volatile String initThreadErrorMessage = "";
    @Setter
    @Getter
    private Timer timer;
    @Setter
    @Getter
    private WorkerTimerTask workerTimerTask;
    @Setter
    @Getter
    private int timerInterval = 2000;
    @Setter
    @Getter
    volatile String pongpongErrorMessage = "";

    @Setter
    @Getter
    public boolean init = false;//zk初始化成功
    @Setter
    @Getter
    public boolean executeTasks = true;
    //
    protected Lock lock = new ReentrantLock();
    //
    @Setter
    @Getter
    private Map<String, List<TaskProcessor>> taskIdToTaskProcessorListMap = new ConcurrentHashMap<String, List<TaskProcessor>>();
    @Setter
    @Getter
    private Map<String, Integer> taskIdToTaskItemVersionMap = new ConcurrentHashMap<String, Integer>();

    @Setter
    @Getter
    private List<String> taskIdList = new ArrayList<String>();

    public ItemTaskWorker() {
        this.ip = Addresss.getLocalHostAddress();
        this.hostName = Addresss.getLocalHostName();
        ApplicationContext applicationContext = ApplicationContextUtils.getApplicationContext();
        Map<String, TaskDataProcessor> stringTaskDataProcessorMap = applicationContext.getBeansOfType(TaskDataProcessor.class);
        for (TaskDataProcessor taskDataProcessor : stringTaskDataProcessorMap.values()) {
            taskIdList.add(taskDataProcessor.getTaskId());
        }
    }

    public ZooKeeperItemTaskDataManager getZooKeeperDataManager() {
        if (this.zooKeeperDataManager == null) {
            throw new NullPointerException("zooKeeperBatchTaskDataManager");
        }
        return zooKeeperDataManager;
    }


    public void init() throws Exception {
        Properties properties = new Properties();
        this.init(properties);
    }

    /**
     * 初始化
     */
    public void init(Properties properties) throws Exception {

        //
        this.lock.lock();
        try {
            if (this.zooKeepersConnectAndWorkerTasksInitThread != null) {
                this.zooKeepersConnectAndWorkerTasksInitThread.stopInitThread();
                this.zooKeepersConnectAndWorkerTasksInitThread = null;
            }
            if (timer != null) {
                timer.cancel();
                timer = null;
            }

            this.zooKeeperDataManager = null;
            if (this.zooKeeperManager != null) {
                this.zooKeeperManager.close();
            }
            //
            Client.setItemTaskWorker(this);
            this.zooKeeperManager = new ZooKeeperManager(properties);
            zooKeepersConnectAndWorkerTasksInitThread = new ZooKeepersConnectAndWorkerTasksInitThread(this);
            zooKeepersConnectAndWorkerTasksInitThread.setName("ItemTaskWorker-zooKeepersConnectAndWorkerTasksInitThread");
            zooKeepersConnectAndWorkerTasksInitThread.start();
        } finally {
            this.lock.unlock();
        }
    }

    /**
     * zk连接成功后进行初始化
     */
    public void initTasksAfterZooKeeperConnected() throws Exception {
        this.zooKeeperManager.doInit();
        this.zooKeeperDataManager = new ZooKeeperItemTaskDataManager(this.zooKeeperManager);
        List<String> taskIdList = this.zooKeeperDataManager.registerWorkerToPermittedTaskList(this);
        log.info("注册成功taskIdList:" + JSON.toJSONString(taskIdList));
        //下面进行worker和zk交互逻辑
        if (this.timer != null) {
            this.timer.cancel();
            this.timer = null;
        }
        this.timer = new Timer("ItemTaskWorker-Timer");
        //
        if (this.workerTimerTask != null) {
            this.workerTimerTask = null;
        }
        this.workerTimerTask = new WorkerTimerTask(this);
        this.timer.schedule(this.workerTimerTask, 2000, this.timerInterval);
    }

    public void pongpong() throws Exception {
        this.lock.lock();
        try {
            Set<String> taskIdSet = this.taskIdToTaskProcessorListMap.keySet();
            for (String taskId : taskIdSet) {
                List<TaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.get(taskId);
                List<TaskProcessor> needRemoveTaskProcessorList = new ArrayList<TaskProcessor>();
                for (TaskProcessor taskProcessor : taskProcessorList) {
                    if (taskProcessor.isExceptionOccurred()) {
                        needRemoveTaskProcessorList.add(taskProcessor);
                    }
                }
                taskProcessorList.removeAll(needRemoveTaskProcessorList);
                for (TaskProcessor taskProcessor : needRemoveTaskProcessorList) {
                    taskProcessor.stop();
                }
            }

        } finally {
            this.lock.unlock();
        }
        this.lock.lock();
        try {
            WorkerInfo workerInfo = null;
            boolean fileNotFoundException = false;
            try {
                workerInfo = this.getZooKeeperDataManager().loadWorkerInfo(this.getWorkerId());//总体信息
            } catch (FileNotFoundException e) {
                log.error("获取workerInfo失败", e);
                fileNotFoundException = true;
            }
            if (fileNotFoundException == true) {//被清理了
                try {
                    stopTaskProcessorList();
                    this.getZooKeeperDataManager().unRegisterWorker(this);//卸载调度器
                    //

                } finally {
                    rebuildTaskProcessorListWhenAllowExecuteTasks();
                }
            } else {
                if (workerInfo.isExecuteTasks() == false) {
                    stopTaskProcessorList();

                } else {
                    //正常
                    rebuildTaskProcessorListWhenAllowExecuteTasks();

                }
            }
        } finally {
            this.lock.unlock();
        }
    }

    /**
     * zk连接不上才重启
     */
    public void reStart() throws Exception {
        try {
            if (this.timer != null) {
                if (this.workerTimerTask != null) {
                    this.workerTimerTask.cancel();
                    this.workerTimerTask = null;//对象并行处理模型
                }
                this.timer.purge();
            }
            //
            this.stopTaskProcessorList();

            //
            if (this.zooKeeperManager != null) {
                this.zooKeeperManager.close();
            }
            this.workerId = null;
            //
            this.init();
        } catch (Throwable e) {
            log.error("" + e.getMessage(), e);
        }
    }

    public void stop() throws Exception {
        try {
            lock.lock();
            this.executeTasks = false;
            if (this.zooKeepersConnectAndWorkerTasksInitThread != null) {
                this.zooKeepersConnectAndWorkerTasksInitThread.stopInitThread();
            }
            if (this.timer != null) {
                if (this.workerTimerTask != null) {
                    this.workerTimerTask.cancel();
                    this.workerTimerTask = null;
                }
                this.timer.cancel();
                this.timer = null;
            }
            this.stopTaskProcessorList();

            if (this.zooKeeperManager != null) {
                this.zooKeeperManager.close();
            }
            if (this.getZooKeeperDataManager() != null) {
                try {
                    ZooKeeper zooKeeper = this.getZooKeeperDataManager().getZooKeeper();
                    if (zooKeeper != null) {
                        zooKeeper.close();
                    }
                } catch (Exception e) {
                    log.error("stopAll zookeeper getZooKeeper", e);
                }
            }
            this.workerId = null;
            log.info("stopAll");
        } catch (Throwable e) {
            log.error("stopAll" + e.getMessage(), e);
        } finally {
            lock.unlock();
        }
    }

    public void rebuildTaskProcessorListWhenAllowExecuteTasks() throws Exception {
        //只支持worker允许运行的情况下才进行调用
        List<String> taskIdList = this.getZooKeeperDataManager().registerWorkerToPermittedTaskListWhenPongPong(this);
        //这个是允许存活的任务,排开这些之外的需要进行清理
        this.stopTaskProcessorListNotTaskIdList(taskIdList);
        this.rebuildTaskProcessorListByWorkerId(taskIdList);
    }


    public TaskProcessor createTaskProcessor(Task task) throws Exception {
        TaskProcessor taskProcessor = null;
        try {
            taskProcessor = new TaskProcessor(task.getTaskId(), this, zooKeeperDataManager, task.getMinThreadCountPerProcessor());
        } catch (Exception e) {
            log.error("schedule" + task.getTaskId(), e);
        }
        return taskProcessor;
    }

    public void stopTaskProcessorList() throws Exception {
        String[] taskIdList = this.taskIdToTaskProcessorListMap.keySet().toArray(new String[0]);
        for (String taskId : taskIdList) {
            for (TaskProcessor taskProcessor : this.taskIdToTaskProcessorListMap.get(taskId)) {
                try {
                    if (taskProcessor != null) {
                        taskProcessor.stop();
                    }
                } catch (Throwable e) {
                    log.error("id=" + taskId, e);
                }
            }
            this.taskIdToTaskProcessorListMap.remove(taskId);
        }
    }

    public void stopTaskProcessorListNotTaskIdList(List<String> taskIdList) throws Exception {
        //找到所有的taskIdList
        List<String> needRemoveTaskIdList = new ArrayList<String>(this.taskIdToTaskProcessorListMap.keySet());
        needRemoveTaskIdList.removeAll(taskIdList);
        for (String taskId : needRemoveTaskIdList) {
            stopTaskProcessorList(taskId);
        }

    }

    public void stopTaskProcessorList(String taskId) throws Exception {
        List<TaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.get(taskId);
        if (taskProcessorList != null) {
            for (TaskProcessor taskProcessor : taskProcessorList) {
                try {
                    if (taskProcessor != null) {
                        taskProcessor.stop();
                    }
                } catch (Throwable e) {
                    log.error("id=" + taskId, e);
                }
            }
            this.taskIdToTaskProcessorListMap.remove(taskId);
        }
    }

    public void rebuildTaskProcessorListByWorkerId(List<String> taskIdList) throws Exception {
        //先进行taskItem状态监听
        for (String taskId : taskIdList) {
            //版本号不一致 则直接清除。如果新的版本号出于分配中则什么也不进行处理。
            TaskItemsInfo taskItemInfo = zooKeeperDataManager.getTaskItemInfo(taskId);
            if (!taskIdToTaskItemVersionMap.containsKey(taskId)) {
                taskIdToTaskItemVersionMap.put(taskId, taskItemInfo.getVersion());
                continue;
            } else {
                int version = taskIdToTaskItemVersionMap.get(taskId);
                if (taskItemInfo.getVersion() > version) {
                    List<TaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.remove(taskId);
                    for (TaskProcessor taskProcessor : taskProcessorList) {
                        try {
                            taskProcessor.stop();//等下个调度时间 会重新根据task能力(因为版本回退可能减少)+worker+ip再重新进行调度器分配。
                        } catch (Throwable e) {
                            log.error(taskId, e);
                        }
                    }
                    taskIdToTaskItemVersionMap.put(taskId, taskItemInfo.getVersion());
                }

            }
        }
        //只要task item 版本号改变。
        //只允许合法的taskIdList
        List<TaskWorkerScheduleInfo> taskWorkerScheduleInfoList = this.getZooKeeperDataManager().loadTaskWorkerScheduleInfoListByWorkerId(this.workerId);
        for (TaskWorkerScheduleInfo taskWorkerScheduleInfo : taskWorkerScheduleInfoList) {
            String taskId = taskWorkerScheduleInfo.getTaskId();
            Task task = getZooKeeperDataManager().loadTask(taskId);
            //
            this.taskIdToTaskProcessorListMap.putIfAbsent(taskId, new ArrayList<TaskProcessor>());
            List<TaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.get(taskId);
            //
            if (!taskIdList.contains(taskId)) {
                //调取不允许该IP或者该worker执行
                for (TaskProcessor taskProcessor : taskProcessorList) {
                    try {
                        taskProcessor.stop();//等下个调度时间 会重新根据task能力(因为版本回退可能减少)+worker+ip再重新进行调度器分配。
                    } catch (Throwable e) {
                        log.error(taskId, e);
                    }
                }
            } else {
                TaskItemsInfo taskItemInfo = zooKeeperDataManager.getTaskItemInfo(taskId);
                if (taskItemInfo.getVersion() > taskWorkerScheduleInfo.getTaskItemVserion()) {
                    for (TaskProcessor taskProcessor : taskProcessorList) {
                        try {
                            taskProcessor.stop();
                        } catch (Throwable e) {
                            log.error(taskId, e);
                        }
                    }
                    //后加
                    taskProcessorList.clear();
                    int requestNum = taskWorkerScheduleInfo.getRequestNum();
                    while (taskProcessorList.size() < requestNum) {
                        TaskProcessor taskProcessor = this.createTaskProcessor(task);
                        taskProcessorList.add(taskProcessor);
                    }
                    continue;//需要重新进行分配。
                } else {
                    //先清理task items过期的处理器
                    for (TaskProcessor taskProcessor : taskProcessorList) {
                        ProcessorInfo processorInfo = zooKeeperDataManager.getProcessorInfo(taskId, taskProcessor.getTaskProcessorInfo().processorId);
                        if (processorInfo == null) {
                            taskProcessorList.remove(taskProcessor);
                            try {
                                taskProcessor.stop();
                            } catch (Throwable e) {
                                log.error(taskId, e);
                            }
                        } else {
                            String needDestroyWhenTaskItemsInvalid = processorInfo.needDestroyWhenTaskItemsInvalid;
                            if ("Y".equals(needDestroyWhenTaskItemsInvalid)) {
                                taskProcessorList.remove(taskProcessor);
                                try {
                                    taskProcessor.stop();
                                } catch (Throwable e) {
                                    log.error(taskId, e);
                                }
                            }
                        }
                    }
                    int requestNum = taskWorkerScheduleInfo.getRequestNum();
                    //先减
                    while (taskProcessorList.size() > 0 && taskProcessorList.size() > requestNum) {
                        TaskProcessor taskProcessor = taskProcessorList.remove(taskProcessorList.size() - 1);
                        try {
                            taskProcessor.stop();
                        } catch (Throwable e) {
                            log.error(taskId, e);
                        }
                    }
                    //调度器线程数分配
                    List<TaskProcessor> newTaskProcessorList = new ArrayList<>();
                    while (taskProcessorList.size() > 0) {
                        TaskProcessor taskProcessor = taskProcessorList.remove(taskProcessorList.size() - 1);
                        int threadCount = taskProcessor.getThreadCountPerTaskProcessor();
                        if (threadCount != task.getMinThreadCountPerProcessor()) {
                            try {
                                taskProcessor.stop();
                            } catch (Throwable e) {
                                log.error(taskId, e);
                            }
                            taskProcessor = this.createTaskProcessor(task);
                        }
                        newTaskProcessorList.add(taskProcessor);
                    }
                    taskProcessorList.addAll(newTaskProcessorList);
                    //后加
                    while (taskProcessorList.size() < requestNum) {
                        TaskProcessor taskProcessor = this.createTaskProcessor(task);
                        taskProcessorList.add(taskProcessor);
                    }
                }
            }

        }
    }


}

